package relop;

import global.RID;
import global.SearchKey;
import heap.HeapFile;
import index.BucketScan;
import index.HashIndex;

/**
 * Implements the hash-based join algorithm described in section 14.4.3 of the
 * textbook (3rd edition; see pages 463 to 464). HashIndex is used to partition
 * the tuples into buckets, and HashTableDup is used to store a partition in
 * memory during the matching phase.
 */
public class HashJoin extends Iterator {

	private BucketScan leftScan;
	private BucketScan rightScan;
	private HeapFile hpFileLeft;
	private HeapFile hpFileRight;
	private HashIndex hpIndexLeft;
	private HashIndex hpIndexRight;

	private Schema schemaLeft;
	private Schema schemaRight;

	private boolean open;

	private boolean newlyCreateLeftFile;
	private boolean newlyCreateRightFile;
	private boolean newlyCreateLeftIndex;
	private boolean newlyCreateRightIndex;

	private HashTableDup hTD;
	private HashTableDup nexthTD;

	private Integer curBucketLeft;
	private Integer curBucketRight;

	private Tuple[] tuplesLeft;
	private Tuple tupleRight;
	private int curTupleArrayIndex;

	private Iterator left;
	private Iterator right;
	private Integer lcol;
	private Integer rcol;
	private Tuple nextTuple;

	/**
	 * Constructs a hash join, given the left and right iterators and which
	 * columns to match (relative to their individual schemas).
	 */
	public HashJoin(Iterator left, Iterator right, Integer lcol, Integer rcol) {
		this.left = left;
		this.right = right;
		this.lcol = lcol;
		this.rcol = rcol;
		schemaLeft = left.schema;
		schemaRight = right.schema;
		this.schema = Schema.join(left.schema, right.schema);

		open = true;

		nextTuple = null;

		hpFileLeft = null;
		hpFileRight = null;
		hpIndexLeft = null;
		hpIndexRight = null;

		curBucketLeft = null;
		curBucketRight = null;

		newlyCreateLeftFile = false;
		newlyCreateRightFile = false;
		newlyCreateLeftIndex = false;
		newlyCreateRightIndex = false;

		leftScan = null;
		rightScan = null;

		hTD = null;
		nexthTD = null;
		tuplesLeft = null;
		this.curTupleArrayIndex = 0;
		tupleRight = null;
		
		partition();
	}

	/**
	 * Gives a one-line explaination of the iterator, repeats the call on any
	 * child iterators, and increases the indent depth along the way.
	 */
	public void explain(int depth) {
		indent(depth);
		String explaination = "Hashjoin on colunms: ";
		explaination += "left col: " + lcol;
		explaination += " and right col: " + rcol;
		System.out.println(explaination);
		left.explain(depth + 1);
		right.explain(depth + 1);
	}

	/**
	 * Restarts the iterator, i.e. as if it were just constructed.
	 */
	public void restart() {
		if (open) {
			leftScan.close();
			rightScan.close();
			leftScan = hpIndexLeft.openScan();
			rightScan = hpIndexRight.openScan();
		}
		curBucketLeft = null;
		curBucketRight = null;
		hTD = null;
		nexthTD = null;

		nextTuple = null;
		tuplesLeft = null;
		tupleRight = null;
		this.curTupleArrayIndex = 0;
	}

	/**
	 * Returns true if the iterator is open; false otherwise.
	 */
	public boolean isOpen() {
		return open;
	}

	/**
	 * Closes the iterator, releasing any resources (i.e. pinned pages).
	 */
	public void close() {
		if (!open) {
			return;
		}

		leftScan.close();
		rightScan.close();
		nextTuple = null;

		left.close();
		right.close();
		left = null;
		right = null;
		leftScan = null;
		rightScan = null;
		hpFileLeft = null;
		hpFileRight = null;
		hpIndexLeft = null;
		hpIndexRight = null;
		curBucketLeft = null;
		curBucketRight = null;
		hTD = null;
		nexthTD = null;
		tuplesLeft = null;
		tupleRight = null;
		this.curTupleArrayIndex = 0;
		open = false;
	}

	private void partition() {
		if (left.getClass().equals(IndexScan.class)) {
			hpIndexLeft = ((IndexScan) left).index;
			hpFileLeft = ((IndexScan) left).file;
	
			leftScan = hpIndexLeft.openScan();
		} else {
			hpIndexLeft = new HashIndex(null);
			newlyCreateLeftIndex = true;
			if (!left.getClass().equals(FileScan.class)) {
				hpFileLeft = new HeapFile(null);
				newlyCreateLeftFile = true;
				while (left.hasNext()) {
					Tuple temp = left.getNext();
					RID ridTemp = hpFileLeft.insertRecord(temp.getData());
					hpIndexLeft.insertEntry(new SearchKey(temp.getField(lcol)),
							ridTemp);
				}

				leftScan = hpIndexLeft.openScan();

			} else {
				FileScan leftFS = (FileScan) left;
				hpFileLeft = leftFS.file;
				while (leftFS.hasNext()) {
					Tuple temp = leftFS.getNext();
					RID ridTemp = leftFS.getLastRID();
					hpIndexLeft.insertEntry(new SearchKey(temp.getField(lcol)),
							ridTemp);
				}
				leftScan = hpIndexLeft.openScan();
		
			}
		}

		if (right.getClass().equals(IndexScan.class)) {
			hpIndexRight = ((IndexScan) right).index;
			hpFileRight = ((IndexScan) right).file;

			rightScan = hpIndexRight.openScan();

		} else {
			hpIndexRight = new HashIndex(null);
			newlyCreateRightIndex = true;
			if (!right.getClass().equals(FileScan.class)) {
				hpFileRight = new HeapFile(null);
				newlyCreateRightFile = true;
				while (right.hasNext()) {
					Tuple temp = right.getNext();
					RID ridTemp = hpFileRight.insertRecord(temp.getData());
					hpIndexRight.insertEntry(
							new SearchKey(temp.getField(rcol)), ridTemp);
				}

				rightScan = hpIndexRight.openScan();

			} else {
				FileScan rightFS = (FileScan) right;
				hpFileRight = rightFS.file;
				while (rightFS.hasNext()) {
					Tuple temp = rightFS.getNext();
					RID ridTemp = rightFS.getLastRID();
					hpIndexRight.insertEntry(
							new SearchKey(temp.getField(rcol)), ridTemp);
				}
				rightScan = hpIndexRight.openScan();

			}

		}

	}

	/**
	 * Returns true if there are more tuples, false otherwise.
	 */
	public boolean hasNext() {
		if (!isOpen()) {
			return false;
		}
		RID tempRidLeft = null, tempRidRight = null;
		Integer tempBucketLeft = null, tempBucketRight = null;

		while (tupleRight == null) {
			if (hTD == null) {
				hTD = nexthTD;
				// If hash table is null, this means the hash table need to be
				// rebuild. First move the two sets to the same bucket.
				while ((curBucketLeft == null) || (curBucketRight == null)
						|| (curBucketLeft.intValue() != curBucketRight.intValue())) {
					if (curBucketLeft == null) {
						if (!leftScan.hasNext()) {
							nextTuple = null;
							return false;
						} else {
							hTD = null;
							nexthTD = null;
							curBucketLeft = leftScan.getNextHash();
							tempRidLeft = leftScan.getNext();
							continue;
						}
					}

					if (curBucketRight == null) {
						if (!rightScan.hasNext()) {
							nextTuple = null;
							return false;
						} else {
							curBucketRight = rightScan.getNextHash();
							tempRidRight = rightScan.getNext();
							continue;
						}
					}

					if (curBucketLeft.intValue() < curBucketRight.intValue()) {
						if (!leftScan.hasNext()) {
							nextTuple = null;
							return false;
						} else {
							hTD = null;
							nexthTD = null;
							curBucketLeft = leftScan.getNextHash();
							tempRidLeft = leftScan.getNext();
							continue;
						}
					}

					if (curBucketLeft.intValue() > curBucketRight.intValue()) {
						if (!rightScan.hasNext()) {
							nextTuple = null;
							return false;
						} else {
							curBucketRight = rightScan.getNextHash();
							tempRidRight = rightScan.getNext();
							continue;
						}
					}
				}

				// We moved to the same bucket. Now build the hash table.
				if (hTD == null) {
					hTD = new HashTableDup();

					Tuple tValueLeft = new Tuple(schemaLeft, hpFileLeft
							.selectRecord(tempRidLeft));
					hTD.add(leftScan.getLastKey(), tValueLeft);
				}
				
				while (leftScan.hasNext()) {
					tempBucketLeft = leftScan.getNextHash();
					tempRidLeft = leftScan.getNext();
					if (curBucketLeft.intValue() != tempBucketLeft) {
						//Store the prefetched tuple in the next hash table
						curBucketLeft = tempBucketLeft;

						Tuple tValueLeft = new Tuple(schemaLeft, hpFileLeft
								.selectRecord(tempRidLeft));
						nexthTD = new HashTableDup();
						nexthTD.add(leftScan.getLastKey(), tValueLeft);
						break;

					}
	
						Tuple tValueLeft = new Tuple(schemaLeft, hpFileLeft
							.selectRecord(tempRidLeft));
						hTD.add(leftScan.getLastKey(), tValueLeft);

				}
			}
			
			// The right tuple may have been prefetched when building the hash table.
			SearchKey rightTupleKey;
	
			if(tempRidRight != null) {
				rightTupleKey = rightScan.getLastKey();
				tuplesLeft = hTD.getAll(rightTupleKey);
			}
			
			// Get the tuples in the hashtable corresponding the the searchkey
			// of the right
			// tuple. Continue Until we found one or the right side move to
			// another bucket.
			while ((tuplesLeft == null) || (tuplesLeft.length == 0)) {
				if (!rightScan.hasNext()) {
					nextTuple = null;
					return false;
				} else {
					tempBucketRight = rightScan.getNextHash();
					tempRidRight = rightScan.getNext();
					if (curBucketRight.intValue() != tempBucketRight) {
						curBucketRight = tempBucketRight;
						hTD = null; // Set to null to trigger hash table rebuild
						break;
					}
					rightTupleKey = rightScan.getLastKey();
					tuplesLeft = hTD.getAll(rightTupleKey);
				}
			}

			if (hTD != null) {

					tupleRight = new Tuple(schemaRight, hpFileRight.selectRecord(tempRidRight));
					this.curTupleArrayIndex = 0;

			}
		}
		
		// At this moment, we could be sure that tuplesLeft is not empty
		// And tupleRight is not empty, too.
		nextTuple = Tuple.join(tuplesLeft[this.curTupleArrayIndex], tupleRight,
				this.schema);
		this.curTupleArrayIndex++;

		if (this.curTupleArrayIndex >= tuplesLeft.length) {
			// Go next right tuple
			tupleRight = null;
			tuplesLeft = null;
		}

		return true;
	}

	/**
	 * Gets the next tuple in the iteration.
	 * 
	 * @throws IllegalStateException
	 *             if no more tuples
	 */
	public Tuple getNext() {
		if (nextTuple == null)
			throw new IllegalStateException("getNext(): no more tuples");

		return nextTuple;
	}

} // public class HashJoin extends Iterator
